Motivation

If a producer is closed when its state is RegisteringSchema, it will not call removeProducer^[1]. Eventually, it leads to a broker-side producer handle leak(Memory leak), and may cause such an error: Producer with name 'xxx' is already connected to topic. The issue will also lead to replicators being stuck.

You can reproduce the issue with the two new tests

• [1]: https://github.com/apache/pulsar/blob/v4.2.1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L1243-L1247

    @Override
    public synchronized CompletableFuture<Void> closeAsync() {
        //...
        //...
        if (cnx == null || currentState != State.Ready) {
            log.info("[{}] [{}] Closed Producer (not connected)", topic, producerName);
            closeAndClearPendingMessages();
            return CompletableFuture.completedFuture(null); // (Highlight) it returns without remove producer from broker-side.
        }
        //...
        //...
}

Modifications

• fix the issue

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment